package com.xxd.dfs.client;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author: XiaoDong.Xie
 * @create: 2020-08-18 10:20
 * @description:
 */
public class FileSystemImpl implements FileSystem {

    private static final Logger logger = LoggerFactory.getLogger(FileSystemImpl.class);

    private NameNodeClient namenode;

    private NIOClient nioClient;

    public FileSystemImpl(NameNodeClient namenode) {
        this.nioClient = new NIOClient();
        this.namenode = namenode;
    }

    /**
     * 创建目录
     *
     * @param path 绝对路劲
     * @return
     */
    @Override
    public boolean mkdir(String path) {
        return namenode.mkdir(path);
    }

    /**
     * 优雅关闭
     */
    @Override
    public void shutdown() {
        namenode.shutdown();
    }

    /**
     * 创建文件
     *
     * @param fileInfo 文件信息
     * @return
     */
    @Override
    public Boolean upload(FileInfo fileInfo) {
        // 在文件目录树种创建文件
        if (!namenode.createFile(fileInfo.getFileName())) {
            logger.info("文件增加到文件目录树失败");
            return false;
        }
        // hadoop 中有文件续约机制，这里省略

        // 请求分配datanode
        String datanodesJson = namenode.allocateDataNodes(fileInfo);
        logger.info("本次namenode 给分配的 两台机器为 " + datanodesJson);

        JSONArray datanodes = JSON.parseArray(datanodesJson);
        for (int i = 0; i < datanodes.size(); i++) {
            Host host = getHost(datanodes.getJSONObject(i));
            nioClient.sendFile(fileInfo, host);
        }


        // 开始往datanode发送文件
        /**
         * hadoop 会先建立连接
         * 然后在发送，发送采用 pipline的方式
         * 我们这里简化一下 直接往两个datanode发送
         * 使用nio 的方式发送数据
         */


        return true;
    }

    /**
     * 解析出Host
     *
     * @param datanode
     * @return
     */
    private Host getHost(JSONObject datanode) {
        Host host = new Host();
        host.setHostname(datanode.getString("hostname"));
        host.setIp(datanode.getString("ip"));
        host.setNioPort(datanode.getInteger("nioPort"));
        return host;
    }
}
